package qdu.edu.com.fushanf4.service.count.usertype.search2characters.count;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import qdu.edu.com.fushanf4.utils.ConfigurationUtil;

import java.io.IOException;
import java.util.*;

// TODO 修改Map的输出及Reduce的输入
public class SearchTwoCharactersCountMapper extends Mapper<LongWritable, Text,Text, IntWritable> {
    private final TreeMap<String, Integer> treeMap = new TreeMap<>();
    private final IntWritable valueOut = new IntWritable();
    private final Text keyOut = new Text();
    private List<String> twoCharactersList;
    private static String twoCharactersPath = null;

    public static void setTwoCharactersPath(String twoCharactersPath) {
        SearchTwoCharactersCountMapper.twoCharactersPath = twoCharactersPath;
    }

    @Override
    protected void setup(Context context) throws IOException {
        if (null != context.getConfiguration().get("twoCharactersPath")) {
            setTwoCharactersPath(context.getConfiguration().get("twoCharactersPath"));
        }
        String fileName = "part-r-00000";
        twoCharactersList = IOUtils.readLines(FileSystem.get(ConfigurationUtil.getConf()).open(new Path(twoCharactersPath, fileName)));
        for (String twoCharacters : twoCharactersList) {
            treeMap.put(twoCharacters, 0);
        }
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) {
        searchKeyValueInMap(treeMap, twoCharactersList, value);
    }

    public static void searchKeyValueInMap (TreeMap<String, Integer> treeMap, List<String> twoCharactersList,Text value) {
        String line = value.toString();
        for (int i = 0; i < treeMap.size(); i++){
            // 此处查找该行是否有该词
            if (line.contains(twoCharactersList.get(i))) {
                treeMap.replace(twoCharactersList.get(i), treeMap.get(twoCharactersList.get(i)), treeMap.get(twoCharactersList.get(i))+1);
            }
        }

    }
    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        for (String value : twoCharactersList) {
            keyOut.set(value);
            valueOut.set(treeMap.get(value));
            context.write(keyOut, valueOut);
        }
    }
}
